package com.nami.dispatcher.domain.event.subscribe;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.thread.ThreadFactoryBuilder;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Multimap;
import com.nami.dispatcher.domain.event.publish.RetryDownMsgOnceEvent;
import com.nami.dispatcher.domain.message.model.RetryMsgRequest;
import com.nami.dispatcher.domain.middleware.mq.DomainEventPublisher;
import com.nami.dispatcher.domain.gateway.UserGateway;
import com.nami.gate.external.basic.cache.user.DeviceInfo;
import com.nami.gate.external.basic.cache.user.UserDeviceRel;
import com.nami.gate.external.basic.constants.DownTopicConst;
import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.dromara.common.core.utils.SessionUtil;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
@RequiredArgsConstructor
public class RetryForwardPackEventListener {

    public static final String SEPARATOR = "&|&";

    /**
     * 能够接收的事件处理 最大时间
     */
    private static final int MAX_WAIT_THRESHOLD = 3;

    private ScheduledExecutorService scheduledExecutorService;

    private Multimap<String, RetryMsgRequest> map = ArrayListMultimap.create();

    private final UserGateway userRepository;
    private final DomainEventPublisher sender;

    @PostConstruct
    public void startup() {
        scheduledExecutorService = new ScheduledThreadPoolExecutor(1
            , ThreadFactoryBuilder.create()
            .setNamePrefix("retry-handler-")
            .setDaemon(true)
            .setUncaughtExceptionHandler(((thread, throwable) ->
                log.error("[retry-handler-task] ThreadPool {} got exception", thread, throwable)))
            .build());
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                doEvict();
            }
        }, 2, 2, TimeUnit.SECONDS);

        log.info("【qos-retry】 消息发送失败兜底策略。");

    }

    /**
     * TODO 如果堆积非常容易出现问题，会出现，用户上线，被踢了。
     */
    private void doEvict() {
        try {
            // 遍历 Multimap.asMap() 并删除键值对
            Iterator<Map.Entry<String, Collection<RetryMsgRequest>>> iterator = map.asMap().entrySet().iterator();
            while (iterator.hasNext()) {
                Map.Entry<String, Collection<RetryMsgRequest>> next = iterator.next();
                String key = next.getKey();
                String appKey = key.split(SEPARATOR)[0];
                String userId = key.split(SEPARATOR)[2];

                List<RetryMsgRequest> retryMsgRequests = (List<RetryMsgRequest>) next.getValue();

                if (retryMsgRequests != null) {
                    // 接收的最新一条数据
                    RetryMsgRequest request = retryMsgRequests.get(retryMsgRequests.size() - 1);
                    //
                    UserDeviceRel qryUser = userRepository.getOnlUserDetailInfoById(appKey, userId);
                    if (qryUser == null) {
                        return;
                    }
//                    MultiResponse<DeviceInfo> onrDevices = userReadFacade.getOnrDevices(appKey, userId);
                    List<DeviceInfo> deviceInfos = qryUser.getDeviceInfos();
                    if (CollUtil.isEmpty(deviceInfos)) {
                        iterator.remove();
                        log.warn("[qos-retry] 重试策略不予发送。当前用户【{}】设备已全部下线 堆积数={}。容器size={}",
                            request.getSessionId(), retryMsgRequests.size(), map.size());
                        return;
                    }
                    String instanceId = SessionUtil.getInstanceId(request.getSessionId());
                    DeviceInfo deviceInfo = deviceInfos.stream()
                        .filter(item -> item.getInstanceId().equals(instanceId))
                        .findFirst()
                        .orElse(null);
                    if (deviceInfo == null) {
                        iterator.remove();
                        log.warn("[qos-retry] 重试策略不予发送。当前用户设备【{}】已下线 当前消息堆积数={} 容器size={}",
                            request.getSessionId(), retryMsgRequests.size(), map.size());
                        return;
                    }

                    // 如果最新的一条数据为五分钟之前数据 则不处理 ，过滤
                    if (DateUtil.between(new Date(request.getCurrTime()), new Date(), DateUnit.MINUTE) > MAX_WAIT_THRESHOLD) {
                        iterator.remove();
                        log.warn("[qos-retry] 超时了重试策略不予发送。堆积数={}", request.getSessionId(), retryMsgRequests.size());
                        return;
                    }
                    sender.publish(DownTopicConst.FLUSH_ONE_TOPIC, request.getMqTags(),  request.getPacket());
                }

                // 删除该键下的所有值
                iterator.remove();
            }
        } catch (Exception var2) {
            var2.printStackTrace();
            log.error("[qos-retry] 事件处理出错了 msg={}", var2.getMessage());
        }

    }

    @EventListener
    public void handleCustomEvent(RetryDownMsgOnceEvent event) {
        RetryMsgRequest request = event.getRequest();
        if (request == null) {
            return;
        }
        map.put(request.getAppKey() + SEPARATOR + SessionUtil.getUserId(request.getSessionId()), request);
    }

}
